package org.common.utils.consumer;
import org.common.utils.thread.ThreadTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 消费者队列
 * @author zhouzhibing
 * @date 2025/8/1
 */
class ConsumerQueue implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(ConsumerQueue.class);

    /**
     * 名称
     */
    private String name;
    /**
     * 线程
     */
    private Thread thread;
    /**
     * 实体队列
     */
    private BlockingQueue<Object> entityQueue;
    /**
     * 执行器
     */
    private Executor executor;
    /**
     * 处理器
     */
    private ConsumerHandler handler;

    /**
     * 是否开始
     */
    private boolean start;

    ConsumerQueue(ConsumerConfig config , int index) {
        this.name = this.getClass().getSimpleName() + "-" + config.getName() + "-" + index;
        this.executor = config.getExecutor();
        this.handler = config.getHandler();
        this.thread = new Thread(this , name);
        this.entityQueue = new LinkedBlockingQueue<>();
    }

    /**
     * 启动
     */
    public void start() {
        start = true;
        thread.start();
    }

    /**
     * 添加异步实体
     * @param entity 异步实体
     */
    public void add(Object entity) {
        if(!start) {
            throw new ConsumerException("ConsumerQueue.add is close , name = " + name + " , entity = " + entity);
        }
        entityQueue.add(entity);
    }

    protected Object take() throws InterruptedException {
        return entityQueue.take();
    }

    @Override
    public void run() {
        while(true) {
            Object currentEntity = null;
            try {
                Object entity = take();
                currentEntity = entity;

                executor.execute(() -> {
                    handler.doHandler(name, entity);
                });

//                if(log.isDebugEnabled()) {
//                    log.debug("ConsumerQueue.run , name = {} , entityQueue.size = {} , currentEntity = {}" , name , entityQueue.size() , currentEntity);
//                }
            } catch (InterruptedException e) {
                if(start) {
                    log.error("ConsumerQueue.run InterruptedException , name = {} , currentEntity = {}" , name , currentEntity , e);
                } else {
                    break;
                }
            } catch (Exception e) {
                log.error("ConsumerQueue.run exception , name = {} , currentEntity = {}" , name , currentEntity , e);
            }
        }
        log.info("ConsumerQueue.run close , name = {}" , name);
    }


    /**
     * 销毁
     */
    protected void destroy() {
        thread = null;
        entityQueue = null;
        handler = null;
    }

    /**
     * 等待处理结束，关闭。
     */
    public void close() {
        start = false;
        while(true) {
            //如果关闭且队列为空
            if(!start && entityQueue.isEmpty()) {
                thread.interrupt();
                destroy();
                break;
            }
            ThreadTool.sleep(100);
        }
    }

    public boolean isStart() {
        return start;
    }

    public String getName() {
        return name;
    }
}